package com.wzy.gmail.realtime.app

import com.wzy.gmail.realtime.util.{MyKafkaUtils, MyOffsetUtils, MyRedisUtils}
import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.kafka.common.TopicPartition
import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.{DStream, InputDStream}
import org.apache.spark.streaming.kafka010.{HasOffsetRanges, OffsetRange}
import org.apache.spark.streaming.{Seconds, StreamingContext}
import com.alibaba.fastjson.{JSON, JSONObject}
import org.apache.spark.broadcast.Broadcast
import redis.clients.jedis.Jedis

import java.util

/**
 * 业务数据分流
 *  1. 准备实时环境
 *     2. 从redis中读取偏移量
 *     3. 从kafka中消费数据
 *     4. 提取偏移量结束点
 *     5. 数据处理
 *     5.1 转换数据结构
 *     5.2 分流
 *     事实数据->kafka
 *     维度数据->Redis
 *     6. flush kafka缓冲区
 *     7. 提交offset
 */
object OdsBaseDbApp {
  def main(args: Array[String]): Unit = {
    // 1. 准备实时环境
    val sparkConf: SparkConf = new SparkConf().setAppName("ods_base_db_app").setMaster("local[4]")
    val ssc = new StreamingContext(sparkConf, Seconds(5))

    val topicName: String = "ODS_BASE_DB_1018"
    val groupId: String = "ODS_BASE_DB_GROUP_1018"

    //2. 从redis中提取偏移量
    val offsets: Map[TopicPartition, Long] = MyOffsetUtils.readOffset(topicName, groupId)

    // 3.从kafka中消费数据
    var kafkaDStream: InputDStream[ConsumerRecord[String, String]] = null
    if (offsets != null && offsets.nonEmpty) {
      kafkaDStream = MyKafkaUtils.getKafkaDStream(topicName, ssc, groupId, offsets)
    } else {
      kafkaDStream = MyKafkaUtils.getKafkaDStream(topicName, ssc, groupId)
    }

    // 4.提取偏移量结束点
    var offsetRanges: Array[OffsetRange] = null
    val offsetRangesDStream: DStream[ConsumerRecord[String, String]] = kafkaDStream.transform(
      rdd => {
        offsetRanges = rdd.asInstanceOf[HasOffsetRanges].offsetRanges
        rdd
      }
    )

    // 5.处理数据
    // 5.1 转换数据结构
    val jsonObjectDStream: DStream[JSONObject] = offsetRangesDStream.map(
      consumerRecord => {
        val dataJson: String = consumerRecord.value()
        val jSONObject: JSONObject = JSON.parseObject(dataJson)
        jSONObject
      }
    )
    //    jsonObjectDStream.print(100)
    // 5.2 分流


    jsonObjectDStream.foreachRDD(
      rdd => {
        //每一个批次数据读一次
        //TODO 如何灵活配置表清单
        // 将表清单维护到redis中，实时任务中动态到redis中获取表清单，
        //类型：string list set zset hash
        // set
        //key: FACT:TABLES DIM:TABLES
        //value: 表名集合
        //写入api: set.add()
        //读取：set.smembers
        //过期： 否
        val redisFactKey: String = "FACT:TABLES"
        val redisDimKey: String = "DIM:TABLES"
        val jedis: Jedis = MyRedisUtils.getJedisFromPool()
        // 事实表清单
        val factTables: util.Set[String] = jedis.smembers(redisFactKey)
        println("事实表清单:" + factTables)
        // 如果driver传给executor数据(数据可序列化)过大，建议用广播变量
        val factTablesBC: Broadcast[util.Set[String]] = ssc.sparkContext.broadcast(factTables)
        // 维度表清单
        val dimTables: util.Set[String] = jedis.smembers(redisDimKey)
        println("维度表清单:" + dimTables)
        // 广播变量
        val dimTablesBC: Broadcast[util.Set[String]] = ssc.sparkContext.broadcast(dimTables)

        jedis.close()

        rdd.foreachPartition(
          jsonObjIter => {
            val jedis: Jedis = MyRedisUtils.getJedisFromPool()
            for (jsonObj <- jsonObjIter) {
              // 提取操作类型
              val operationType: String = jsonObj.getString("type")
              val opValue: String = operationType match {
                case "bootstrap-insert" => "I"
                case "insert" => "I"
                case "update" => "U"
                case "delete" => "D"
                case _ => null
              }
              // 判断操作类型：1.明确什么操作 2.过滤不感兴趣数据
              if (opValue != null) {
                // 提取表名
                val tableName: String = jsonObj.getString("table")

                if (factTablesBC.value.contains(tableName)) {
                  //事实数据
                  // 提取数据
                  val data: String = jsonObj.getString("data")
                  val dwdTopicName: String = s"DWD_${tableName.toUpperCase}_${opValue}_1018"
                  MyKafkaUtils.send(dwdTopicName, data)
                }
                if (dimTablesBC.value.contains(tableName)) {
                  //维度数据
                  //类型list string set zset hash
                  //    整个表存成一个hash需要考虑表的数据量大小和访问频次
                  //    hash:一条数据存为一个hash
                  //    String：一条数据保存成一个jsonString
                  //key： DIM+表名+id
                  //value：整条数据JsonString
                  //写入api：set
                  //读取api：get
                  //过期：否

                  // 提取id
                  val dataObj: JSONObject = jsonObj.getJSONObject("data")
                  val id: String = dataObj.getString("id")
                  val redisKey: String = s"DIM:${tableName.toUpperCase}:${id}"

                  jedis.set(redisKey, dataObj.toJSONString)
                }
              }
            }
            jedis.close()

            //刷新kafka缓冲区
            MyKafkaUtils.flush()
          }
        )
        //提交offset
        MyOffsetUtils.saveOffset(topicName, groupId, offsetRanges)
      }
    )

    ssc.start()
    ssc.awaitTermination()
  }

}
